package com.xrui.hbase;

import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.xrui.hbase.exception.InvalidNameException;

import com.xrui.hbase.filter.ColumnFilter;
import com.xrui.hbase.schema.spec.ColumnReaderSpec;

import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * <p>Builds a request for columns of data to read from a HBase table.</p>
 * <p>
 * <p>You can instantiate a DataRequestBuilder using the {@link
 * DataRequest#builder()} method.</p>
 * <p>
 * <p>{@link DataRequest} objects are immutable; this object helps you construct
 * them. With a DataRequest builder, you can set various properties that affect
 * the data request as a whole (for example, the timestamp interval to retrieve).
 * You can also use the {@link #newColumnsDef()} method of this object to create instances of
 * {@link DataRequestBuilder.ColumnsDef}, which allow you to define a set of columns that
 * are part of a data request, and their retrieval properties.</p>
 * <p>
 * <p>A DataRequestBuilder.ColumnsDef object has two types of methods: methods
 * starting with <code>with...</code> define properties associated with some columns. The
 * <code>add()</code> methods then attach specific columns to the data request being built,
 * using the properties previously specified. You may call <code>add()</code> or
 * <code>addFamily()</code> multiple times.
 * You may not overwrite the value of a property like maxVersions once it's already been set.</p>
 * <p>
 * <p>It is an error to change properties within a column request (e.g., call {@link
 * DataRequestBuilder.ColumnsDef#withMaxVersions(int)}) after using the
 * <code>add()</code> or <code>addFamily()</code> methods to add columns to the request.</p>
 * <p>
 * <p>The following behaviors are errors and are not allowed:</p>
 * <ul>
 * <li>Adding the same column multiple times. This includes definitions like
 * <tt>info:name</tt> that conflict with requests for the entire
 * <tt>info:*</tt> family.</li>
 * <li>Calling a property-setting method (<tt>withTimeRange()</tt>,
 * <tt>withMaxVersions()</tt>, etc) more than once on a given
 * <tt>DataRequestBuilder</tt> or {@link DataRequestBuilder.ColumnsDef}.
 * These methods will throw IllegalStateException.</li>
 * <li>Changing any properties after calling build() to construct the DataRequest.</li>
 * <li>Calling build() more than once.</li>
 * </ul>
 * <p>
 * <h2>Usage example</h2>
 * <p>
 * <p>For the common case of reading <tt>info:foo</tt>, <tt>info:bar</tt>,
 * <tt>info:baz</tt>, and <tt>products:*</tt>:</p>
 * <pre>
 * DataRequestBuilder builder = DataRequest.builder().withTimeRange(t1, t2);
 * builder.newColumnsDef().withMaxVersions(42)
 *     .add("info", "foo")
 *     .add("info", "bar")
 *     .add("info", "baz")
 *     .addFamily("products");
 * DataRequest request = builder.build();
 * </pre>
 * <p>
 * This can also be written as:
 * <pre>
 * final DataRequest request = DataRequest.build()
 *     .addColumns(ColumnsDef.create()
 *         .withMaxVerions(42)
 *         .add("info", "foo")
 *         .add("info", "bar")
 *         .add("info", "baz")
 *         .addFamily("products"))
 *     .build();
 * </pre>
 * <p>
 * <p>To add <tt>fam1:col1</tt>, <tt>fam1:col2</tt>, and <tt>fam2:*</tt>, each with
 * different retrieval properties to the same request, do the following:</p>
 * <pre>
 * DataRequestBuilder builder = DataRequest.builder().withTimeRange(t1, t2);
 * builder.newColumnsDef().withMaxVersions(10).withPageSize(p).add("fam1", "col1");
 * builder.newColumnsDef().add("fam1", "col2");
 * builder.newColumnsDef().withMaxVersions(42).addFamily("fam2");
 * DataRequest request = builder.build();
 * </pre>
 */
public final class DataRequestBuilder {

    /**
     * Column builders associated with this data request builder.
     * <p>
     * <p>We use a linked set so we construct DataRequests in a stable order.
     * This is necessary for testing HBaseDataRequestAdapter.</p>
     */
    private LinkedHashSet<ColumnsDef> mColumnsDefs = Sets.newLinkedHashSet();

    /**
     * The minimum timestamp of cells to be read (inclusive).
     */
    private long mMinTimestamp = Constants.BEGINNING_OF_TIME;

    /**
     * The maximum timestamp of cells to be read (exclusive).
     */
    private long mMaxTimestamp = Constants.END_OF_TIME;

    /**
     * True if the user already set timestamp range.
     */
    private boolean mIsTimeRangeSet;

    /**
     * True if we already built an object.
     */
    private boolean mIsBuilt = false;

    // -----------------------------------------------------------------------------------------------

    /**
     * Constructor. Package-private; use {@link DataRequest#builder()} to create an
     * instance of this.
     */
    DataRequestBuilder() {
    }

    // -----------------------------------------------------------------------------------------------

    /**
     * Sets the time range of cells to return: [<code>minTimestamp</code>,
     * <code>maxTimestamp</code>).
     *
     * @param minTimestamp Request cells with a timestamp at least minTimestamp.
     * @param maxTimestamp Request cells with a timestamp less than maxTimestamp.
     * @return This data request builder instance.
     */
    public DataRequestBuilder withTimeRange(long minTimestamp, long maxTimestamp) {
        checkNotBuilt();
        Preconditions.checkArgument(minTimestamp >= 0,
            "minTimestamp must be positive or zero, but got: %d", minTimestamp);
        Preconditions.checkArgument(maxTimestamp > minTimestamp,
            "Invalid time range [%d--%d]", minTimestamp, maxTimestamp);
        Preconditions.checkState(!mIsTimeRangeSet,
            "Cannot set time range more than once.");

        mIsTimeRangeSet = true;
        mMinTimestamp = minTimestamp;
        mMaxTimestamp = maxTimestamp;
        return this;
    }

    /**
     * Return a builder for columns associated with this DataRequestBuilder.
     * <p>
     * <p>Creates an object that allows you to specify a set of related columns attached
     * to the same DataRequest that all share the same retrieval properties, like
     * the number of max versions.</p>
     *
     * @return a new DataRequestBuilder.ColumnsDef builder object associated with this
     * data request builder.
     */
    public ColumnsDef newColumnsDef() {
        checkNotBuilt();
        final ColumnsDef c = new ColumnsDef();
        mColumnsDefs.add(c);
        return c;
    }

    /**
     * Return a builder for columns, initialized from an existing
     * {@link DataRequest.Column}.
     * <p>
     * <p>Creates an object that allows you to specify a set of related columns attached
     * to the same DataRequest that all share the same retrieval properties, like
     * the number of max versions.</p>
     * <p>
     * <p>This builder will have all properties fully initialized, and it will already
     * include a request for the column named as an argument. Only additional calls to
     * <code>DataRequestBuilder.ColumnsDef.add(...)</code> are permitted.</p>
     *
     * @param existingColumn is a Column from an existing DataRequest object that should
     *                       be included in this new DataRequest.
     * @return a new DataRequestBuilder.ColumnsDef builder object associated with this
     * data request builder.
     */
    public ColumnsDef newColumnsDef(DataRequest.Column existingColumn) {
        return newColumnsDef()
            .withFilter(existingColumn.getFilter())
            .withPageSize(existingColumn.getPageSize())
            .withMaxVersions(existingColumn.getMaxVersions())
            .add(existingColumn.getFamily(), existingColumn.getQualifier());
    }

    /**
     * Adds another set of column definitions to this DataRequest builder.
     * <p>
     * <p>Columns added in this manner must not redefine any column definitions already included in
     * the DataRequestBuilder. It is an error to add a ColumnsDef instance to multiple
     * DataRequestBuilders.
     *
     * @param def A set of column definitions contained in a {@link DataRequestBuilder.ColumnsDef}
     *            instance.
     * @return this DataRequest builder.
     */
    public DataRequestBuilder addColumns(ColumnsDef def) {
        def.seal();
        mColumnsDefs.add(def);
        return this;
    }

    /**
     * Construct a new DataRequest based on the configuration specified in this builder
     * and its associated column builders.
     * <p>
     * <p>After calling build(), you may not use the builder anymore.</p>
     *
     * @return a new DataRequest object containing the column requests associated
     * with this DataRequestBuilder.
     */
    public DataRequest build() {
        checkNotBuilt();
        mIsBuilt = true;

        // Entire families for which a definition has been recorded:
        final Set<String> families = Sets.newHashSet();

        // Fully-qualified columns for which a definition has been recorded:
        final Set<ColumnName> columns = Sets.newHashSet();

        // Families of fully-qualified columns for which definitions have been recorded:
        final Set<String> familiesOfColumns = Sets.newHashSet();

        final List<DataRequest.Column> requestedColumns = Lists.newArrayList();
        // Iterate over the ColumnsDefs in the order they were added (mColumnsDef is a LinkedHashSet).
        for (ColumnsDef columnsDef : mColumnsDefs) {
            for (DataRequest.Column column : columnsDef.buildColumns()) {
                if (column.getQualifier() == null) {
                    final boolean isNotDuplicate = families.add(column.getFamily());
                    Preconditions.checkState(isNotDuplicate,
                        "Duplicate definition for family '%s'.", column.getFamily());

                    Preconditions.checkState(!familiesOfColumns.contains(column.getFamily()),
                        "DataRequest may not simultaneously contain definitions for family '%s' "
                            + "and definitions for fully qualified columns in family '%s'.",
                        column.getFamily(), column.getFamily());

                } else {
                    final boolean isNotDuplicate = columns.add(column.getColumnName());
                    Preconditions.checkState(isNotDuplicate, "Duplicate definition for column '%s'.", column);

                    Preconditions.checkState(!families.contains(column.getFamily()),
                        "DataRequest may not simultaneously contain definitions for family '%s' "
                            + "and definitions for fully qualified columns '%s'.",
                        column.getFamily(), column.getColumnName());
                    familiesOfColumns.add(column.getFamily());
                }
                requestedColumns.add(column);
            }
        }
        return new DataRequest(requestedColumns, mMinTimestamp, mMaxTimestamp);
    }

    /**
     * @throws IllegalStateException after the DataRequest has been built with {@link #build()}.
     *                               Prevents reusing this builder.
     */
    private void checkNotBuilt() {
        Preconditions.checkState(!mIsBuilt,
            "DataRequest builder cannot be used after build() is invoked.");
    }

    /**
     * Defines properties associated with one or more columns in a request for HBase table columns.
     * <p>
     * <p>See {@link DataRequestBuilder} for a larger specification of how
     * {@link DataRequest} objects are constructed.</p>
     * <p>
     * <p>Use the {@link DataRequestBuilder#newColumnsDef()} method to create an instance of
     * a column set definition. This object has no "build()" method; use the {@link
     * DataRequestBuilder#build()} method to build the entire {@link
     * DataRequest} at once.</p>
     * <p>
     * <p>It is an error to call one of the <tt>with...()</tt> methods after calling
     * <tt>add()</tt> to add a specific column definition. You must define all
     * properties of the columns before specifying particular columns to attach to
     * the DataRequest.</p>
     * <p>
     * <p>It is an error to request a column more than once in the same
     * DataRequest.</p>
     */
    public static final class ColumnsDef {
        /**
         * Becomes true when the columns definition is sealed.
         */
        private boolean mSealed = false;

        /**
         * The maximum number of versions from the column to read (of the most recent).
         */
        private Integer mMaxVersions = null;

        /**
         * A column filter (may be null).
         */
        private ColumnFilter mFilter = null;

        /**
         * Becomes true once the filter is set.
         */
        private boolean mFilterInitialized = false;

        /**
         * The number of cells per page (zero means no paging).
         */
        private Integer mPageSize;

        /**
         * Columns in this definition.
         */
        private List<ColumnName> mColumns = Lists.newArrayList();

        /**
         * Optional per-column specification of read properties used to decode cells.
         */
        private Map<ColumnName, ColumnReaderSpec> mColumnReaderSpec = Maps.newHashMap();

        /**
         * Creates a new requested <code>ColumnsDef</code> builder.
         */
        private ColumnsDef() {
        }

        /**
         * @return a new builder for column definitions.
         */
        public static ColumnsDef create() {
            return new ColumnsDef();
        }

        /**
         * @return true if the user has already started assigning columns to this builder.
         */
        private boolean assignedColumns() {
            return !mColumns.isEmpty();
        }

        /**
         * If the user has assigned columns to this builder, throw IllegalStateException.
         */
        private void checkNoCols() {
            Preconditions.checkState(!mSealed,
                "ColumnsDef cannot be used once DataRequestBuilder.build() had been called.");
            Preconditions.checkState(!assignedColumns(),
                "Properties of the columns builder cannot be changed once columns are assigned to it.");
        }

        /**
         * Sets the maximum number of the most recent versions of a particular column to return.
         * <p>
         * <p> For a map type column family, the maximum number of versions applies to each qualifier
         * individually.  In particular, the maximum number of versions does not limit the total
         * number of qualifiers or of versions returned for the entire family.
         * </p>
         * <p>
         * <p> Note: the maximum number of versions is currently not working properly when using
         * a {@link Pager} to page through a map-type family.
         * In particular, when the maximum number of versions is greater than the page size,
         * the total number of cells returned for a qualifier may exceed the configured maximum.
         * </p>
         *
         * @param maxVersions The maximum number of versions per column qualifier to return.
         * @return This column request builder instance.
         */
        public ColumnsDef withMaxVersions(int maxVersions) {
            checkNoCols();
            Preconditions.checkState(mMaxVersions == null,
                "Cannot set max versions to %s, max versions already set to %s.",
                maxVersions, mMaxVersions);
            Preconditions.checkArgument(maxVersions > 0,
                "Maximum number of versions must be strictly positive, but got: %s",
                maxVersions);

            mMaxVersions = maxVersions;
            return this;
        }

        /**
         * Sets a filter to attach to each column specified by this column request builder.
         *
         * @param filter The column filter;
         * @return This column request builder instance.
         */
        public ColumnsDef withFilter(ColumnFilter filter) {
            checkNoCols();
            Preconditions.checkState(!mFilterInitialized, "Cannot set filter multiple times");

            mFilter = filter;
            mFilterInitialized = true;
            return this;
        }

        /**
         * Configures paging and sets the page size (i.e. the maximum number of cells per page).
         * <p>
         * <p>
         * The page size is an upper limit to the number of cells retrieved from the region servers
         * at a time, to bound the amount of memory consumed on the client machine.
         * See {@link Pager} for more details.
         * </p>
         * <p> Defaults to zero, which means paging is disabled. </p>
         *
         * @param pageSize The maximum number of cells to return in each page of results.
         *                 Use 0 to disable paging and return all results at once.
         * @return This column request instance.
         */
        public ColumnsDef withPageSize(int pageSize) {
            checkNoCols();
            Preconditions.checkState(mPageSize == null,
                "Cannot set page size to %s, page size already set to %s.", pageSize, mPageSize);
            Preconditions.checkArgument(pageSize >= 0,
                "Page size must be 0 (disabled) or positive, but got: %s", mPageSize);

            mPageSize = pageSize;
            return this;
        }

        /**
         * Adds a column to the data request, using the properties associated with this
         * DataRequestBuilder.ColumnsDef object.
         * <p>
         * <p>
         * Data request properties cannot be modified (using <code>withX()</code> methods)
         * after columns have been added to the column request (using <code>add()</code> methods).
         * </p>
         *
         * @param family the column family to retrieve as a map.
         * @return this column request builder instance.
         * @throws InvalidNameException if the family name is invalid.
         */
        public ColumnsDef addFamily(final String family) {
            return add(ColumnName.create(family, null));
        }

        /**
         * Requests data from a map-type family, and specifies read properties for that family.
         * <p>
         * <p>
         * Data request properties cannot be modified (using <code>withX()</code> methods)
         * after columns have been added to the column request (using <code>add()</code> methods).
         * </p>
         *
         * @param family           Map-type family to request data from.
         * @param columnReaderSpec Read properties to resolve when decoding cells from the column family.
         * @return this column request builder instance.
         * @throws InvalidNameException if the family name is invalid.
         */
        public ColumnsDef addFamily(
            final String family,
            final ColumnReaderSpec columnReaderSpec
        ) {
            return add(ColumnName.create(family, null), columnReaderSpec);
        }

        /**
         * Adds a column to the data request, using the properties associated with this
         * DataRequestBuilder.ColumnsDef object.
         * <p>
         * <p>
         * Data request properties cannot be modified (using <code>withX()</code> methods)
         * after columns have been added to the column request (using <code>add()</code> methods).
         * </p>
         *
         * @param family    the column family of the column to retrieve.
         * @param qualifier the qualifier of the column to retrieve.
         * @return this column request builder instance.
         * @throws InvalidNameException if the column name is invalid.
         */
        public ColumnsDef add(
            final String family,
            final String qualifier
        ) {
            return add(ColumnName.create(family, qualifier));
        }

        /**
         * Requests data from a column and specifies read properties for that column.
         * <p>
         * <p>
         * Data request properties cannot be modified (using <code>withX()</code> methods)
         * after columns have been added to the column request (using <code>add()</code> methods).
         * </p>
         *
         * @param family           Family of the column to request data from.
         * @param qualifier        Qualifier of the column to request data from.
         *                         Null means request an entire map-type family.
         * @param columnReaderSpec Read properties to resolve when decoding cells from the column.
         * @return this column request builder instance.
         * @throws InvalidNameException if the column name is invalid.
         */
        public ColumnsDef add(
            final String family,
            final String qualifier,
            final ColumnReaderSpec columnReaderSpec
        ) {
            return add(ColumnName.create(family, qualifier), columnReaderSpec);
        }

        /**
         * Adds a column to the data request, using the properties associated with this
         * DataRequestBuilder.ColumnsDef object.
         * <p>
         * <p>
         * Data request properties cannot be modified (using <code>withX()</code> methods)
         * after columns have been added to the column request (using <code>add()</code> methods).
         * </p>
         *
         * @param column the column name to retrieve.
         * @return this column request builder instance.
         */
        public ColumnsDef add(final ColumnName column) {
            return add(column, null);
        }

        /**
         * Adds a column to the data request, using the properties associated with this
         * DataRequestBuilder.ColumnsDef object.
         * <p>
         * <p>
         * Data request properties cannot be modified (using <code>withX()</code> methods)
         * after columns have been added to the column request (using <code>add()</code> methods).
         * </p>
         *
         * @param column           the column name to retrieve.
         * @param columnReaderSpec Specification of read properties used when decoding cells.
         * @return this column request builder instance.
         */
        public ColumnsDef add(
            final ColumnName column,
            final ColumnReaderSpec columnReaderSpec
        ) {
            Preconditions.checkState(!mSealed,
                "Cannot add more columns to this ColumnsDef after build() has been called.");
            Preconditions.checkArgument(!mColumnReaderSpec.containsKey(column),
                "Duplicate request for column '%s'.", column);
            mColumns.add(column);
            mColumnReaderSpec.put(column, columnReaderSpec);
            return this;
        }

        /**
         * Seals this columns definition.
         * <p>
         * The definition becomes immutable, and the only method call allowed from there on is build().
         */
        private void seal() {
            Preconditions.checkState(!mSealed, "ColumnsDef is sealed.  This usually indicates that the "
                + "ColumnsDef has been used in the construction of another DataRequest already.");
            mSealed = true;
        }

        /**
         * Builds the columns associated with this column builder.
         *
         * @return a list of constructed DataRequest.Column objects associated with the
         * output DataRequest instance.
         */
        private List<DataRequest.Column> buildColumns() {
            if (!mSealed) {
                mSealed = true;
            }

            // Values not previously initialized are now set to default values.
            // This builder is immutable after this method is called, so this is ok.
            if (mPageSize == null) {
                mPageSize = DataRequest.PAGING_DISABLED;
            }

            if (mMaxVersions == null) {
                mMaxVersions = 1;
            }

            final List<DataRequest.Column> columns = Lists.newArrayListWithCapacity(mColumns.size());
            for (ColumnName column : mColumns) {
                columns.add(new DataRequest.Column(
                    column.getFamily(),
                    column.getQualifier(),
                    mMaxVersions,
                    mFilter,
                    mPageSize,
                    mColumnReaderSpec.get(column)
                ));
            }
            return columns;
        }
    }
}
